package cover;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.regex.Pattern;

/**
 * Created by zhaopeng on 2/14/17.
 */
public class CoverMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text key = new Text();
    private Text value = new Text();
    private String[] uidAndToken = new String[2];

    @Override
    public void map(LongWritable keys, Text values, Context context) throws IOException, InterruptedException {
        String line = values.toString();
        String[] args = line.split(Pattern.quote("\t"));

        if (1 == args.length) {
            if (getUidAndToken(line)) {
                key.set(uidAndToken[1]);
                value.set("token:" + uidAndToken[0]);
            } else {
                return;
            }
        } else if (2 == args.length) {
            key.set(args[0]);
            value.set("true");

        } else {
            return;
        }
        context.write(key,value);
    }


    private boolean getUidAndToken(String line) {
        String[] args = line.split(Pattern.quote("^"));
        int counter = 0;
        for (String param : args) {
            String[] tmp = param.split("=");
            if (2 != tmp.length)
                continue;
            if ("token".equals(tmp[0])) {
                uidAndToken[0] = tmp[1];
                counter++;
            } else if ("uuid".equals(tmp[0])) {
                uidAndToken[1] = tmp[1];
                if(uidAndToken[1].length()==32||uidAndToken[1].length()==36)
                    uidAndToken[1]= uidAndToken[1].toUpperCase();
                counter++;
            }
        }

        return counter == 2;
    }
}
